package service

import (
	"context"
	"encoding/base64"
	"fmt"
	"git.multiverse.io/eventkit/common/model/transaction"
	"git.multiverse.io/eventkit/common/util"
	"git.multiverse.io/eventkit/dxc-server/config"
	"git.multiverse.io/eventkit/kit/client/mesh"
	"git.multiverse.io/eventkit/kit/codec"
	codec_text "git.multiverse.io/eventkit/kit/codec/text"
	frameConsts "git.multiverse.io/framework/common/constant"
	"strconv"

	dxcConstant "git.multiverse.io/eventkit/dxc-server/constant"
	"git.multiverse.io/eventkit/dxc-server/mgmt/metrics"
	"git.multiverse.io/eventkit/dxc-server/models"
	"git.multiverse.io/eventkit/dxc-server/repo"
	"git.multiverse.io/eventkit/kit/client"
	"git.multiverse.io/eventkit/kit/constant"
	"git.multiverse.io/eventkit/kit/contexts"
	"git.multiverse.io/framework/log"
	jsoniter "github.com/json-iterator/go"
	"github.com/pkg/errors"
	"runtime/debug"
	"strings"
	"sync"
	"time"
)

var (
	ChannelPool     chan int
	ResponseChannel []chan error
	once            sync.Once
)

type CallbackPair struct {
	Task *util.Future
	T    *models.Transaction
}

type CallbackFail struct {
	branch *models.Transaction
	err    error
}

var SomeBranchCallbackFailed = errors.New("Some branch callback failed")

var json = jsoniter.ConfigCompatibleWithStandardLibrary

type CallbackService interface {
	TryFailedIgnoreCallbackCancel(rootXid string, branchXid string) (int, error)
	GlobalTxnCallback(spanCtx *contexts.SpanContexts, rootXid string, isCallbackConfirm bool) (int, int, error)
	BranchTxnCallback(spanCtx *contexts.SpanContexts, t *models.Transaction, isCallbackConfirm bool) (err error)
}

// @Desc new a CallbackService
func NewCallbackService(txnRepository repo.Repository, client client.Client, ctx context.Context) CallbackService {
	once.Do(func() {
		ChannelPool = make(chan int, config.Cfg.Service.MaxCallbackChannel)
		ResponseChannel = make([]chan error, config.Cfg.Service.MaxCallbackChannel)
		for index := 0; index < config.Cfg.Service.MaxCallbackChannel; index++ {
			ChannelPool <- index
			ResponseChannel[index] = make(chan error)
		}
	})

	return &DefaultCallbackService{
		Repository: txnRepository,
		Client:     client,
		Ctx:        ctx,
	}
}

// @Desc define a DefaultCallbackService
type DefaultCallbackService struct {
	Repository repo.Repository
	Client     client.Client
	Ctx        context.Context
}

// @Desc call back branch confirm method
// @Param spanCtx span context
// @Param Transaction info
// @Return err error
func (s *DefaultCallbackService) BranchTxnCallback(spanCtx *contexts.SpanContexts, t *models.Transaction, isCallbackConfirm bool) (err error) {
	var callbackType string
	if isCallbackConfirm {
		callbackType = "confirm"
	} else {
		callbackType = "cancel"
	}
	startTime := time.Now()
	var callbackServiceName string
	var request interface{}
	// for metrics
	if config.Cfg.Metrics.Enable {
		if isCallbackConfirm {
			metrics.DTSCallbackConfirmTPSAnalyser.IncreaseCount()
			defer func() {
				endTime := time.Now()
				timeCostMicroseconds := endTime.Sub(startTime).Microseconds()
				metrics.DTSCallbackConfirmTimeCostAvgAnalyser.IncreaseTotal(float64(timeCostMicroseconds))
				if nil != err {
					metrics.DTSCallbackConfirmSuccessRateAnalyser.IncreaseCount(false)
				} else {
					metrics.DTSCallbackConfirmSuccessRateAnalyser.IncreaseCount(true)
				}
			}()
		} else {
			metrics.DTSCallbackCancelTPSAnalyser.IncreaseCount()
			defer func() {
				endTime := time.Now()
				timeCostMicroseconds := endTime.Sub(startTime).Microseconds()
				metrics.DTSCallbackCancelTimeCostAvgAnalyser.IncreaseTotal(float64(timeCostMicroseconds))
				if nil != err {
					metrics.DTSCallbackCancelSuccessRateAnalyser.IncreaseCount(false)
				} else {
					metrics.DTSCallbackCancelSuccessRateAnalyser.IncreaseCount(true)
				}
			}()
		}
	}

	if isCallbackConfirm {
		callbackServiceName = "BranchTxnConfirm"
	} else {
		if t.TryFailedIgnoreCallbackCancel {
			log.Debugsf("Callback branch Cancel: this branch not need callback cancel, Branch[%++v] ", t)
			return nil
		}
		callbackServiceName = "BranchTxnCancel"
	}

	request = transaction.AtomicTxnCallbackRequest{
		Head: transaction.TxnEventHeader{
			Service: callbackServiceName,
		},
		Request: transaction.AtomicTxnCallbackRequestBody{
			RootXid:     t.RootXid,
			ParentXid:   t.ParentXid,
			BranchXid:   t.BranchXid,
			RequestTime: util.CurrentTime(),
			ServiceName: t.ServiceName,
			Headers:     t.Headers,
		},
	}

	var e error
	var reqBytes []byte
	if t.IsClientSDKVersion(constant.ClientSDKVersionV2) {
		paramBytes, err := base64.StdEncoding.DecodeString(t.ParamData) // base64 decode param
		if err != nil {
			err = fmt.Errorf("callback branch %s: Encode Request Param base64 failed, (RootXid: %s,  Branch:%s), err:%s", callbackType, t.RootXid, t.BranchXid, e)
			return err
		}
		reqBytes, e = util.Encode(request, paramBytes)
	} else {
		reqBytes, e = json.Marshal(request)
	}

	if e != nil {
		err = fmt.Errorf("callback branch %s: Encode Request failed, (RootXid: %s,  Branch:%s), err:%s", callbackType, t.RootXid, t.BranchXid, e)
		return err
	}

	// direct mode
	var resBody []byte

	participantAddressArray := strings.Split(t.ParticipantAddress, constant.ParticipantAddressSplitChar)

	if isCallbackConfirm && 0 == len(participantAddressArray[2]) {
		log.Debugsf("callback branch confirm: doesn't have confirm method, Skip branch confirm:%++v", t)
		return nil
	} else if !isCallbackConfirm && 0 == len(participantAddressArray[3]) {
		log.Debugsf("callback branch cancel: doesn't have cancel method, Skip branch cancel:%++v", t)
		return nil
	}
	log.Debugsf("Callback branch %s: Request[%++v], timeout:[%d]", callbackType, request, spanCtx.TimeoutMilliseconds)
	if resBody, err = s.doCallback(spanCtx, reqBytes, participantAddressArray, isCallbackConfirm, t); nil != err {
		t.BranchTxnLastUpdateTime = time.Now()
		err = fmt.Errorf("callback branch %s: callback failed, (RootXid: %s,  Branch:%s), err:%s", callbackType, t.RootXid, t.BranchXid, err)
		return err
	}

	endTime := time.Now()
	commonResp := &models.CommonResponse{}
	if e := json.Unmarshal(resBody, commonResp); e != nil {
		err = fmt.Errorf("callback branch %s: Decode Response failed, (RootXid: %s,  Branch:%s), err:%s, respBody:%s", callbackType, t.RootXid, t.BranchXid, e, string(resBody))
		t.BranchTxnLastUpdateTime = time.Now()
		return err
	}

	log.Debugsf("Callback branch %s: Response[%++v]", callbackType, commonResp)

	success, err := commonResp.CheckSuccess()
	if err != nil {
		err = fmt.Errorf("callback branch %s: commonResp CheckSuccess failed, (RootXid: %s,  Branch:%s), err:%s", callbackType, t.RootXid, t.BranchXid, err)
		t.BranchTxnLastUpdateTime = time.Now()
		return err
	}

	if !success {
		t.BranchTxnLastUpdateTime = time.Now()
		err = fmt.Errorf("callback branch %s: response failed, (RootXid: %s, Branch:%s), Response:%++v", callbackType, t.RootXid, t.BranchXid, commonResp)
		return err
	}

	if isCallbackConfirm {
		t.BranchTxnStat = dxcConstant.BRANCH_TXN_STATUS_CONFIRMED
	} else {
		t.BranchTxnStat = dxcConstant.BRANCH_TXN_STATUS_CANCELLED
	}

	t.BranchTxnLastUpdateTime = time.Now()
	log.Debugsf("Callback branch %s: Success transaction[%++v], Cost:[%v]", callbackType, t, endTime.Sub(startTime))

	return nil
}

// @Desc call back all branches
// @Param spanCtx span context
// @Param rootXid transaction's rootXid
// @Param isCallbackConfirm means It's try to confirm or cancel
// @Return total transactions && success callback transaction
// @Return err error
func (s *DefaultCallbackService) GlobalTxnCallback(spanCtx *contexts.SpanContexts, rootXid string, isCallbackConfirm bool) (int, int, error) {
	var callbackType string
	var root *models.RootTransaction
	var err error
	if isCallbackConfirm {
		callbackType = "confirm"
	} else {
		callbackType = "cancel"
	}
	// 0:trying, 1-confirming, 2-confirmed, 3-cancelling, 4-cancelled
	if isCallbackConfirm {
		root, err = s.Repository.TryCallback(rootXid, dxcConstant.ROOT_TXN_STATUS_CONFIRMING)
	} else {
		// 0:trying, 1-confirming, 2-confirmed, 3-cancelling, 4-cancelled
		root, err = s.Repository.TryCallback(rootXid, dxcConstant.ROOT_TXN_STATUS_CANCELLING)
	}
	if err != nil {
		return 0, 0, err
	}
	numberOfBranches := len(root.Branches)

	// for metrics
	if config.Cfg.Metrics.Enable {
		metrics.DTSParticipantAvgAnalyser.IncreaseTotal(float64(numberOfBranches))
	}

	timeoutMilliseconds := spanCtx.TimeoutMilliseconds - config.Cfg.Service.MaxServiceConsumeMilliseconds
	if timeoutMilliseconds < 0 {
		timeoutMilliseconds = spanCtx.TimeoutMilliseconds
	}
	spanCtx.ParentSpanID = spanCtx.SpanID

	tasks := make([]*CallbackPair, 0)
	for _, branch := range root.Branches {
		if branch.IsRoot() && branch.IsClientSDKVersion(constant.ClientSDKVersionV1) {
			// The current version of client is V1, If the branch is root then confirm/cancel will execute by client SDK
			log.Infosf("Skip %s root transaction[%s]!", callbackType, rootXid)
			continue
		}
		index := <-ChannelPool
		future := util.NewFuture()
		tasks = append(tasks, &CallbackPair{future, branch})
		go func(index int, branch *models.Transaction) {
			defer func() {
				if err := recover(); err != nil {
					catchError := fmt.Errorf("callback branch catch error:[%++v], stack=[%s]", err, string(debug.Stack()))
					future.SetResult(catchError)
				}
				ChannelPool <- index
			}()
			future.SetResult(s.BranchTxnCallback(spanCtx, branch, isCallbackConfirm))
		}(index, branch)
	}
	success := make([]*models.Transaction, 0)
	fails := make([]*CallbackFail, 0)
	for _, callbackPair := range tasks {
		result := callbackPair.Task.GetResult(timeoutMilliseconds + 2000)
		if result == nil {
			success = append(success, callbackPair.T)
		} else {
			var e error
			switch result.(type) {
			case error:
				e = result.(error)
			default:
				e = fmt.Errorf("unknow response, result=%++v", result)
			}
			callbackFail := CallbackFail{
				branch: callbackPair.T,
				err:    e,
			}
			fails = append(fails, &callbackFail)
		}
	}
	if len(fails) != 0 {
		_, err = s.Repository.CallbackFail(rootXid, root.Branches, false)
		if err != nil {
			err := errors.Errorf("CallbackFail fail RootXid:[%s]} failed, error:[%v]", rootXid, err)
			return len(tasks), len(success), err
		}
		// send root xid to abnormal txn handler
		s.Repository.ExpireTime(rootXid)
		log.Debugsf("some branch callback fail[%s], need recover, rootXid:%s", callbackType, rootXid)
		traceId := ""
		for _, branch := range root.Branches {
			if branch.IsRoot() {
				traceId = branch.ParentXid
			}
		}
		spanCtx.TimeoutMilliseconds = 30000
		spanCtx.ParentSpanID = rootXid
		spanCtx.TraceID = traceId
		pair := models.Pair{
			RootXid:  rootXid,
			SpanCtx:  spanCtx,
			RootType: models.CallbackFailedRecordsRecover,
			Type:     models.CallbackFailedRecordsRecover,
		}
		models.Queue.Push(pair)
		// build error msg
		branchMsgs := make([]string, 0)
		for _, fail := range fails {
			branchMsg := fmt.Sprintf("service:%s, failMsg:[%++v]", fail.branch.ServiceName, fail.err)
			branchMsgs = append(branchMsgs, branchMsg)
		}
		return len(tasks), len(success), errors.Errorf("Some branch callback failed, err:{%++v}", branchMsgs)
	}
	// all callback success
	if _, err := s.Repository.DeleteTxn(rootXid); err != nil {
		err := errors.Errorf("Remove root transaction RootXid:[%s]  failed, error:[%v]", rootXid, err)
		return len(tasks), len(success), err
	}
	return len(tasks), len(success), nil
}

func (s *DefaultCallbackService) TryFailedIgnoreCallbackCancel(rootXid string, branchXid string) (int, error) {
	return s.Repository.UpdateBranchTryFailedIgnoreCallbackCancel(rootXid, branchXid)
}

func (s *DefaultCallbackService) doCallback(spanCtx *contexts.SpanContexts, reqBytes []byte, participantAddressArray []string, isCallbackConfirm bool, t *models.Transaction) (resBody []byte, err error) {
	timeoutMilliseconds := spanCtx.TimeoutMilliseconds - config.Cfg.Service.MaxServiceConsumeMilliseconds
	if timeoutMilliseconds < 0 {
		timeoutMilliseconds = spanCtx.TimeoutMilliseconds
	}
	header := make(map[string]string)
	if isCallbackConfirm {
		header[dxcConstant.APM_ATTACH] = dxcConstant.APM_ATTACH_CALLBACK_CONFIRM
	} else {
		header[dxcConstant.APM_ATTACH] = dxcConstant.APM_ATTACH_CALLBACK_CANCEL
	}
	// set to3
	header[frameConsts.To3] = strconv.Itoa(timeoutMilliseconds)
	requestOptions := []client.RequestOption{
		mesh.WithMaxRetryTimes(0),
		mesh.WithTimeout(time.Duration(timeoutMilliseconds) * time.Millisecond),
		mesh.WithCodec(codec.BuildCustomCodec(&codec_text.Encoder{}, &codec_text.Decoder{})),
		mesh.WithHeader(header),
	}
	if strings.EqualFold(constant.CommDirect, participantAddressArray[0]) {
		/* comm type + "|" + Callback root url + "|" + confirm path + "|" + cancel path */
		callbackURL := ""
		if isCallbackConfirm {
			callbackURL = participantAddressArray[1] + participantAddressArray[2]
		} else {
			callbackURL = participantAddressArray[1] + participantAddressArray[3]
		}
		requestOptions = append(requestOptions, []client.RequestOption{
			mesh.WithHTTPRequestInfo(
				callbackURL,
				frameConsts.DefaultHTTPMethodPost,
				"",
			),
		}...)
	} else {
		/* comm type + "|" + Org + "/" + "WKS" + "/" + "ENV" + "/" + "SU" + "/" + "NodeId" + "/" + "InstanceId" + "|" + confirm topicId + "|" + cancel topicId*/
		var eventId string

		topicAttrArray := strings.Split(participantAddressArray[1], constant.TopicIDSplitChat)

		if isCallbackConfirm {
			eventId = participantAddressArray[2]

		} else {
			eventId = participantAddressArray[3]
		}

		requestOptions = append(requestOptions, []client.RequestOption{
			mesh.WithORG(topicAttrArray[0]),
			mesh.WithWorkspace(topicAttrArray[1]),
			mesh.WithEnvironment(topicAttrArray[2]),
			mesh.WithSU(topicAttrArray[3]),
			mesh.WithNodeID(topicAttrArray[4]),
			mesh.WithInstanceID(topicAttrArray[5]),
			mesh.WithEventID(eventId),
		}...)
	}

	meshRequest := mesh.NewMeshRequest(reqBytes)
	meshRequest.WithOptions(requestOptions...)
	if config.Cfg.Service.IgnoreCheckClientSDKVersion {
		if config.Cfg.Service.IgnoreClientSDKVersionWithDxc {
			meshRequest.WithOptions(mesh.WithTopicTypeDxc())
		} else {
			meshRequest.WithOptions(mesh.WithTopicTypeDts())
		}
	} else {
		if t.IsClientSDKVersion(constant.ClientSDKVersionV2) {
			meshRequest.WithOptions(mesh.WithTopicTypeDxc())
		} else {
			meshRequest.WithOptions(mesh.WithTopicTypeDts())
		}
	}
	_, err = s.Client.SyncCall(s.Ctx, meshRequest, &resBody)

	return resBody, err
}
